/*
 * SrsRtuStreamCaChe.cpp
 *
 *  Created on: 2017年6月16日
 *      Author: liutingting
 */
#include "stdafx.h"
#include "SrsRtuStreamCaChe.h"

#define SRS_RTU_MEMORY_MANGER_SLICE_COUNT 50
#define SRS_RTU_MEMORY_RECOVERY_QUEUE_SIZE_NUM 200
#define SRS_RTU_MIN_RTNS_AUDIO_PACKET_LOSS_RATE 0.0001
#define SRS_RTU_SECONDS_TO_OPEN_CACHE_TIME 5000
#define SRS_RTU_CACHE_SIZE_RESERVED_TIME  100

#define srs_freep(p) \
    if (p) { \
        delete p; \
        p = 0; \
    } \
    (void)0


// please use the freepa(T[]) to free an array,
// or the behavior is undefined.
#define srs_freepa(pa) \
    if (pa) { \
        delete[] pa; \
        pa = 0; \
    } \
    (void)0


long long rtu_get_system_time()
{
    srs_update_system_time_ms();
    return srs_get_system_time_ms();
}

long long rtu_get_system_us()
{
    timeval now;

    //if (gettimeofday(&now, 0) < 0) {
        return -1;
    //}

    // @see: https://github.com/ossrs/srs/issues/35
    // we must convert the tv_sec/tv_usec to long long.
    long long now_us = ((long long)now.tv_sec) * 1000 * 1000 + (long long)now.tv_usec;
    return now_us;

}


SrsRtuSliceNoMap::SrsRtuSliceNoMap(int init_count)
{
    slice_sum = 0;
    insert_slice_num = 0;
    insert_payload_size = 0;

    pkts = new SrsRtuStreamDataPacket*[init_count];
	memset(pkts, 0, sizeof(void*)* init_count);
}

SrsRtuSliceNoMap::SrsRtuSliceNoMap(long long time_now, int sum, DWORD first_packet_no, DWORD message_no, DWORD timestamp, int id)
{
    slice_sum = sum;
    insert_slice_num = 0;
    insert_payload_size = 0;
    _start_time = time_now;

    pkts = new SrsRtuStreamDataPacket*[slice_sum];
	memset(pkts, 0, sizeof(void*)* slice_sum);


    head_packetno = first_packet_no;
    last_packetno = first_packet_no + sum -1;

    _timestamp = timestamp;
    _message_no = message_no;
}

SrsRtuSliceNoMap::~SrsRtuSliceNoMap()
{
    clear();

    srs_freepa(pkts);
}

int SrsRtuSliceNoMap::initialize(int sum, DWORD first_packet_no, DWORD message_no, DWORD timestamp, int id)
{
    slice_sum = sum;

    // increase vector.
    if (slice_sum >= SRS_RTU_MEMORY_MANGER_SLICE_COUNT) {

        // use new array.
        srs_freepa(pkts);

        pkts = new SrsRtuStreamDataPacket*[slice_sum];
		memset(pkts, 0, sizeof(void*)* slice_sum);
    }

    insert_slice_num = 0;
    insert_payload_size = 0;

    head_packetno = first_packet_no;
    last_packetno = first_packet_no + sum -1;

    _timestamp = timestamp;
    _message_no = message_no;

	return 0;
}

long long SrsRtuSliceNoMap::start_time()
{
    return _start_time;
}

bool SrsRtuSliceNoMap::insert(SrsRtuStreamDataPacket* packet)
{
    DWORD slice_no = packet->slice_no;

    if(pkts[slice_no] == 0) {
        pkts[slice_no] = packet;
        insert_slice_num++;
        insert_payload_size += packet->get_size();

        return true;

    } else {
        srs_freep(packet);
        return false;
    }

}

int SrsRtuSliceNoMap::get_size()
{
    return insert_slice_num;
}

int SrsRtuSliceNoMap::get_insert_payload_size()
{
    return insert_payload_size;
}

bool SrsRtuSliceNoMap::is_existing(int slice_no)
{
    return pkts[slice_no] != 0;
}

void SrsRtuSliceNoMap::clear()
{
    for(int i = 0; i < slice_sum; i++) {
        srs_freep(pkts[i]);
    }

    slice_sum = 0;

    insert_slice_num = 0;
    insert_payload_size = 0;
}

int SrsRtuSliceNoMap::rbegin_packetno()
{
    int last_packet_no = -1;

    for(int i = insert_slice_num - 1; i >= 0; i--) {
        if(pkts[i] != 0) {
            last_packet_no = pkts[i]->packet_no;
            break;
        }
    }

    return last_packet_no;
}



bool SrsRtuSliceNoMap::check_is_complete()
{
    if(slice_sum == insert_slice_num)
    {
        return true;
    }
    return false;
}

int  SrsRtuSliceNoMap::get_loss_packet_num()
{
    return slice_sum - insert_slice_num;
}

int SrsRtuSliceNoMap::get_slice_sum()
{
    return slice_sum;
}


int SrsRtuSliceNoMap::get_head_packetno()
{
    return head_packetno;
}

int SrsRtuSliceNoMap::get_last_packetno()
{
    return last_packetno;
}

DWORD SrsRtuSliceNoMap::get_messageno()
{
    return _message_no;
}

long long  SrsRtuSliceNoMap::get_timestamp()
{
    return _timestamp;
}

SrsRtuStreamDataPacket *SrsRtuSliceNoMap::get_packet(int slice_no)
{
    if(slice_no > slice_sum) {
        return 0;
    }

    return pkts[slice_no];
}

/**
 * @brief The SrsRtuMessageNoMap class
 *   the key is message_no  , the value SrsRtuSliceNoMap.
 */

SrsRtuMessageNoMap::SrsRtuMessageNoMap()
{
    max_packet_no = 0;
    min_packet_no = 0;
    insert_payload_size = 0;

    old_messageno = 0;
    video_min_packetno = 0;
    audio_min_packetno = 0;

}
SrsRtuMessageNoMap::~SrsRtuMessageNoMap()
{
    clear();
}

bool SrsRtuMessageNoMap::push(SrsRtuStreamDataPacket* packet)
{
//    srs_freep(packet);
//    return true;

//    packetno_map.insert(std::pair <DWORD, SrsRtuStreamDataPacket*>(packet->packet_no, packet));

//    if(packetno_map.size() >= 100)
//    {
//        std::map<DWORD, SrsRtuStreamDataPacket*>::iterator it;

//        for(it = packetno_map.begin(); it != packetno_map.end();)
//        {
//            SrsRtuStreamDataPacket* packet = it->second;
//            srs_freep(packet);
//            it++;
//        }
//        packetno_map.clear();
//    }
//    return true;

//    srs_freep(packet);
//    return true;

    std::map<DWORD, SrsRtuSliceNoMap*>::iterator it;
    it = messageno_map.find(packet->message_no);

//    srs_freep(packet);
//    return true;

    bool flag = false;
    if(it == messageno_map.end())
    {
        long long time_now = srs_get_system_time_ms();
        SrsRtuSliceNoMap* sliceno_queue = new SrsRtuSliceNoMap(time_now, packet->slice_sum, packet->packet_no - packet->slice_no,
                                                               packet->message_no, packet->timestamp, packet->stream_id);
        messageno_map[packet->message_no] = sliceno_queue;
        flag = sliceno_queue->insert(packet);
    }
    else
    {
        flag = it->second->insert(packet);
    }

    if(flag)
    {
       insert_payload_size += packet->payload_size;

       if(packet->payload_size < 0 || packet->payload_size > 1300)
       {
       }

       if(max_packet_no < packet->packet_no || max_packet_no <= 0 )
       {
          max_packet_no = packet->packet_no;
       }
       if(min_packet_no > packet->packet_no || min_packet_no <= 0)
       {
           min_packet_no = packet->packet_no;
       }

       //insert packet_no key
//       if(packet->packet_no > SRS_RTU_FIRST_PACKET_NO)
//       {
//          packetno_map.insert(std::pair <DWORD, SrsRtuStreamDataPacket*>(packet->packet_no, packet));
//       }

       //insert packet_no key
             if(packet->packet_no > SRS_RTU_FIRST_PACKET_NO)
             {
                //packetno_map.insert(std::pair <DWORD, SrsRtuStreamDataPacket*>(packet->packet_no, packet));

                if(packet->stream_type == SrsRtuStreamDataTypeVideo
                        || packet->stream_type == SrsRtuStreamDataTypeVideoKeyframe)
                {
                    if(video_min_packetno > packet->video_packet_no || video_min_packetno <= 0)
                    {
                        video_min_packetno = packet->video_packet_no;
                    }

                    video_rtns_map.insert(std::pair <DWORD, SrsRtuStreamDataPacket*>(packet->video_packet_no, packet));
                }
                else if(packet->stream_type == SrsRtuStreamDataTypeAudio)
                {
                    if(audio_min_packetno > packet->audio_packet_no || audio_min_packetno <= 0)
                    {
                        audio_min_packetno = packet->audio_packet_no;
                    }

                    audio_rtns_map.insert(std::pair <DWORD, SrsRtuStreamDataPacket*>(packet->audio_packet_no, packet));
                }
             }


    }

    return flag;
}

SrsRtuSliceNoMap* SrsRtuMessageNoMap::pop()
{
    std::map<DWORD, SrsRtuSliceNoMap*>::iterator it;
    it = messageno_map.begin();
    if(it == messageno_map.end())
    {
        return 0;
    }

    SrsRtuSliceNoMap* slieceno_map = it->second;
    erase(it);

    return slieceno_map;
}


int SrsRtuMessageNoMap::get_size()
{
    return messageno_map.size();
}

int SrsRtuMessageNoMap::get_insert_payload_size()
{
    return insert_payload_size;
}

message_no_iterator SrsRtuMessageNoMap::begin()
{
    return messageno_map.begin();
}

message_no_iterator SrsRtuMessageNoMap::end()
{
    return messageno_map.end();
}

void SrsRtuMessageNoMap::erase_begin()
{
    std::map<DWORD, SrsRtuSliceNoMap*>::iterator it= messageno_map.begin();
    erase(it);
}

void SrsRtuMessageNoMap::erase(message_no_iterator iter)
{
    if(iter->first > SRS_RTU_FIRST_MESSAGE_NO)
    {
        if(iter->second->get_last_packetno() != -1)
        {
            //update the min_packet_no
            min_packet_no = iter->second->get_last_packetno();
            last_timestamp = iter->second->get_timestamp() ;
        }

        int last_packet_no = iter->second->rbegin_packetno();
        erase_packetno_map(last_packet_no);
    }

    insert_payload_size -= iter->second->get_insert_payload_size();
    old_messageno = iter->first;
    messageno_map.erase(iter);


    //update the min_packet_no
//    std::map<DWORD, SrsRtuSliceNoMap*>::iterator it;
//    it = messageno_map.begin();
//    if(it != messageno_map.end())
//    {
//        min_packet_no = it->second->get_head_packetno();
//    }

}


void SrsRtuMessageNoMap::erase_packetno_map(DWORD last_packet_no)
{
    std::map<DWORD, SrsRtuStreamDataPacket*>::iterator it;

//    for(it = packetno_map.begin(); it != packetno_map.end();)
//    {
//        if(it->first > last_packet_no)
//        {
//            break;
//        }
//
//        packetno_map.erase(it++);
//    }
//

    for(it = video_rtns_map.begin(); it != video_rtns_map.end();)
     {
         //if(it->first > last_packet_no)
         if(it->second->packet_no > last_packet_no)
         {
             break;
         }

         video_rtns_map.erase(it++);
     }

     for(it = audio_rtns_map.begin(); it != audio_rtns_map.end();)
     {
         //if(it->first > last_packet_no)
         if(it->second->packet_no > last_packet_no)
         {
             break;
         }

         audio_rtns_map.erase(it++);
     }


}

void SrsRtuMessageNoMap::erase_video_packetno_map(DWORD last_packet_no)
{
    std::map<DWORD, SrsRtuStreamDataPacket*>::iterator it;

    for(it = video_rtns_map.begin(); it != video_rtns_map.end();)
    {
        if(it->first > last_packet_no)
        {
            break;
        }

        video_rtns_map.erase(it++);
    }
}

void SrsRtuMessageNoMap::erase_audio_packetno_map(DWORD last_packet_no)
{
    std::map<DWORD, SrsRtuStreamDataPacket*>::iterator it;

    for(it = audio_rtns_map.begin(); it != audio_rtns_map.end();)
    {
        if(it->first > last_packet_no)
        {
            break;
        }

        audio_rtns_map.erase(it++);
    }
}

void SrsRtuMessageNoMap::clear()
{
    std::map<DWORD, SrsRtuSliceNoMap*>::iterator it;

    for(it = messageno_map.begin(); it != messageno_map.end();)
    {
        SrsRtuSliceNoMap* sliceno_map = it->second;
        srs_freep(sliceno_map);
        it++;
    }
    messageno_map.clear();
    //packetno_map.clear();
    video_rtns_map.clear();
    audio_rtns_map.clear();

    insert_payload_size = 0;
}

void SrsRtuMessageNoMap::get_loss_packetno_list(std::vector<DWORD> &packetno_list)
{
    std::map<DWORD, SrsRtuStreamDataPacket*>::iterator it;
    DWORD next_packet_no = 0;
    for(it = packetno_map.begin(); it != packetno_map.end();)
    {

        if(it == packetno_map.begin())
        {
            //printf("min packetno = %d\n", min_packet_no);

            if(it->first - min_packet_no > 1)
            {
                for(DWORD i = min_packet_no + 1; i < it->first; i++)
                {
                    packetno_list.push_back(i);
                }
            }

        }

        if(next_packet_no != 0 && next_packet_no != it->first)
        {
            for(DWORD i = next_packet_no; i < it->first; i++)
            {
                packetno_list.push_back(i);
                //rtu_trace("rtns request packetno = %d, min_packet_no=%d, max_packet_no=%d---------------", i, min_packet_no, max_packet_no);
            }
        }

        next_packet_no = it->first + 1;

        it++;
    }
}

void SrsRtuMessageNoMap::get_loss_packetno_list(std::vector<DWORD> &video_packetno_list, std::vector<DWORD> &audio_packetno_list)
{
    //calculate video loss packetno , insert to list
    std::map<DWORD, SrsRtuStreamDataPacket*>::iterator it;
    DWORD next_packet_no = 0;
    for(it = video_rtns_map.begin(); it != video_rtns_map.end();)
    {
        if(it == video_rtns_map.begin())
        {
            if(it->first - video_min_packetno > 1)
            {
                for(DWORD i = video_min_packetno + 1; i < it->first; i++)
                {
                    video_packetno_list.push_back(i);
                    //rtu_trace("video rtns request packetno = %d, video_min_packetno = %d, size = %d, video_rtns_map_head_packetno = %d",
                      //        i, video_min_packetno, video_rtns_map.size(), it->first);
                }
            }

        }

        if(next_packet_no != 0 && next_packet_no != it->first)
        {
            for(DWORD i = next_packet_no; i < it->first; i++)
            {
                video_packetno_list.push_back(i);
                //rtu_trace("video rtns request packetno = %d, video_min_packetno = %d---------------", i, video_min_packetno );
            }
        }

        next_packet_no = it->first + 1;

        it++;
    }


    //calculate audio loss packetno , insert to list
    next_packet_no = 0;
    for(it = audio_rtns_map.begin(); it != audio_rtns_map.end();)
    {
        if(it == audio_rtns_map.begin())
        {
            if(it->first - audio_min_packetno > 1)
            {
                for(DWORD i = audio_min_packetno + 1; i < it->first; i++)
                {
                    audio_packetno_list.push_back(i);
//                    rtu_trace("audio rtns request packetno = %d---------------", i);
                }
            }

        }

        if(next_packet_no != 0 && next_packet_no != it->first)
        {
            for(DWORD i = next_packet_no; i < it->first; i++)
            {
                audio_packetno_list.push_back(i);
//                rtu_trace("audio rtns request packetno = %d, audio_min_packetno = %d---------------", i, audio_min_packetno);
            }
        }

        next_packet_no = it->first + 1;

        it++;
    }
}


DWORD SrsRtuMessageNoMap::get_min_packetno()
{
    //return min_packet_no;
//    std::map<DWORD, SrsRtuStreamDataPacket*>::iterator it;
//    it = packetno_map.begin();
//    if(it != packetno_map.end())
//    {
//        if(it->second->slice_no != 0)
//        {
//            min_packet_no = it->first - it->second->slice_no;
//        }
//        else
//        {
//             min_packet_no = it->first;
//        }
//    }

    return min_packet_no;
}

int SrsRtuMessageNoMap::get_max_packetno()
{
    return max_packet_no;
}

int SrsRtuMessageNoMap::get_packet_num()
{
    return max_packet_no - min_packet_no;
}

bool SrsRtuMessageNoMap::check_is_complete()
{
    std::map<DWORD, SrsRtuSliceNoMap*>::iterator it;
    it = messageno_map.begin();
    if(it == messageno_map.end())
    {
        return false;
    }
    return it->second->check_is_complete();
}

bool SrsRtuMessageNoMap::check_is_complete(int message_no)
{
    std::map<DWORD, SrsRtuSliceNoMap*>::iterator it;
    it = messageno_map.find(message_no);
    if(it == messageno_map.end())
    {
        return false;
    }
    return it->second->check_is_complete();
}

DWORD SrsRtuMessageNoMap::get_cache_time()
{
    DWORD cache_time = 0;
//    if(packetno_map.size() > 0)
//    {
//        cache_time = packetno_map.rbegin()->second->timestamp - packetno_map.begin()->second->timestamp;
//    }

    if(messageno_map.size() > 0)
     {
         //cache_time = packetno_map.rbegin()->second->timestamp - packetno_map.begin()->second->timestamp;
         cache_time = messageno_map.rbegin()->second->get_timestamp() - messageno_map.begin()->second->get_timestamp();
     }

//    if(messageno_map.size() >= 150)
//    {
//        cache_time = 800;
//    }
    return cache_time;
}

long long SrsRtuMessageNoMap::get_message_interval_timestamp()
{
    std::map<DWORD, SrsRtuSliceNoMap*>::iterator it = messageno_map.begin();
    return ( it->second->get_timestamp() - last_timestamp - 5);
}

bool SrsRtuMessageNoMap::get_is_output_flag()
{
    DWORD messageno =  messageno_map.begin()->first;
    if( (old_messageno + 1) == messageno || old_messageno == 0)
    {
        return true;
    }

    return false;
}

int SrsRtuMessageNoMap::get_next_message_packetno(int message_no)
{
    std::map<DWORD, SrsRtuSliceNoMap*>::iterator it;
    it = messageno_map.find(message_no);
    if(it != messageno_map.end())
    {
        return it->second->get_last_packetno();
    }
    return -1;
}

int  SrsRtuMessageNoMap::memory_recovery(SrsRtuSliceNoMap* sliceno_map)
{
	return 0;
}





SrsRtuStreamCaChe::SrsRtuStreamCaChe()
{
    cache_sh = new SrsRtuMessageNoMap();
    cache_media =  new SrsRtuMessageNoMap();

    first_play_time = 0;
    cache_min_size = 0;

    start_statistics_time = 0;
    rtns_start_time = 0;
    max_packetno = 0;
    min_packetno = 0;
    recv_packet_num = 0;
    rtns_packet_num = 0;
    sh_num = 0;

    cache_time = 300;  //ms
    cache_stream_size = 200*1024/8; //*300/1000;  //kbps
    rtns_check_period = 20; //ms
    rtns_level = 1;
    round_trip_time = 0;
    calculate_rtns_level_period = 0;
    cache_output_time = 20;
    cache_output_waittime = 5;
    update_output_time_flag = true;
    wait_cache_flag = true;

    nagle_cache_time = 50; //ms
    nagle_cache_size = 200*1024/8/2;

    packet_loss_rate = 0.0;
    rtns_packet_loss_rate = 0.0;

    check_sh_full_flag = true;
    sh_full_flag = false;
    process_packet_flag = false;

    insert_rtmp_total_size = 0;
    insert_rtmp_second_size = 0;
    del_rtmp_second_size = 0;
    statistics_time = packet_check_time = output_timeout = force_output_time = cache_no_data_time = rtu_get_system_time();

    last_audio_packetno = 0;
    loss_audio_packet_count = 0;
    max_audio_packetno = 0;
    min_audio_packetno = 0;
    max_video_packetno = 0;
}

SrsRtuStreamCaChe::~SrsRtuStreamCaChe()
{
    srs_freep(cache_sh);
    srs_freep(cache_media);
}


void SrsRtuStreamCaChe::set_url(std::string _url)
{
     url = _url;
}

void SrsRtuStreamCaChe::set_stream_id(int _stream_id)
{
     stream_id = _stream_id;
}

int SrsRtuStreamCaChe::get_cache_size()
{
    return cache_media->get_size();
}

void SrsRtuStreamCaChe::packet_push(SrsRtuStreamDataPacket* packet)
{

    bool insert_flag = false;
    if(packet->message_no < SRS_RTU_FIRST_MESSAGE_NO)  //should be define
    {
        insert_flag = cache_sh->push(packet);
        if(cache_sh->get_size() == 3)
        {
            sh_full_flag = true;
            check_sh_full_flag = false;
        }
    }
    else
    {
        if(packet->packet_no > cache_media->get_min_packetno() )
        {
            insert_flag = cache_media->push(packet);
        }
        else
        {
            srs_freep(packet);
        }
    }

    if(insert_flag)
    {
       recv_packet_num++;
       calculate_max_and_min_packetno(packet);
    }
    calculate_packet_loss_rate();


 //   rtu_to_rtmp();
}

void SrsRtuStreamCaChe::rtns_packet_push(SrsRtuStreamDataPacket* packet)
{
    //srs_trace("rtns push packetno=%d", packet->packet_no);

    bool insert_flag = false;
    DWORD packet_no = packet->packet_no;
    if(packet->message_no < SRS_RTU_FIRST_MESSAGE_NO)
    {
        insert_flag = cache_sh->push(packet);

        if(cache_sh->get_size() == 3)
        {
            sh_full_flag = true;
            check_sh_full_flag = false;
        }
    }
    else
    {
        if(packet->packet_no > cache_media->get_min_packetno() )
        {
            insert_flag = cache_media->push(packet);
        }
        else
        {
            srs_freep(packet);
        }
    }
    if(insert_flag)
    {
        rtns_packet_num++;
        calculate_max_and_min_packetno(packet);
    }

    calculate_rtns_packet_loss_rate();
   // rtu_to_rtmp();
}

void SrsRtuStreamCaChe::get_loss_packetno_list(std::vector<DWORD> &packetno_list)
{
    int ret = ERROR_SUCCESS;
    if(check_sh_full_flag)
    {
//        if(cache_sh->get_size() == 3)
//        {
//            sh_full_flag = true;
//            check_sh_full_flag = false;
//        }
//        else
//        {
//            if( (ret = process_rtns_sh_packet(packetno_list)) != ERROR_SUCCESS)
//            {
//                rtu_error("process rtns sh packet error.");
//            }
//        }

        if(cache_sh->get_size() < 3)
        {
            if( (ret = process_rtns_sh_packet(packetno_list)) != ERROR_SUCCESS)
            {
            }
        }


    }

    if(cache_media->get_size() > 0)
    {
        if( (ret = process_rtns_stream_packet(packetno_list)) != ERROR_SUCCESS)
        {
            //break;
        }

    }
}

void SrsRtuStreamCaChe::get_loss_packetno_list(std::vector<DWORD> &video_packetno_list, std::vector<DWORD> &audio_packetno_list)
{
    int ret = ERROR_SUCCESS;
    if(check_sh_full_flag)
    {
        if(cache_sh->get_size() < 3)
        {
            if( (ret = process_rtns_sh_packet(video_packetno_list)) != ERROR_SUCCESS)
            {
              //  rtu_error("process rtns sh packet error.");
            }
        }
    }

    if(cache_media->get_size() > 0)
    {
        if( (ret = process_rtns_stream_packet(video_packetno_list, audio_packetno_list)) != ERROR_SUCCESS)
        {
          //  rtu_error("process rtns stream packet error.");
            //break;
        }

    }
}

int SrsRtuStreamCaChe::process_rtns_sh_packet(std::vector<DWORD> &packetno_list)
{
    int ret =ERROR_SUCCESS;
    srs_update_system_time_ms();
    long long time_now = srs_get_system_time_ms();
    if(time_now - sh_check_time >= 30)
    {
        //srs_trace("rtns sh packet");
        std::map<DWORD, SrsRtuSliceNoMap*>::iterator it;
        it = cache_sh->begin();
        for(DWORD i =1; i <= 3; i++)
        {
            if(it == cache_sh->end())
            {
                packetno_list.push_back(i);
                continue;
            }
            else
            {
                if(i == it->first)
                {
                    it++;
                }
                else
                {
                    packetno_list.push_back(i);
                }
            }
        }
        sh_check_time = time_now;
    }
    return ret;
}

int SrsRtuStreamCaChe::process_rtns_stream_packet(std::vector<DWORD> &packetno_list)
{
    int ret = ERROR_SUCCESS;
    srs_update_system_time_ms();
    long long time_now = srs_get_system_time_ms();
    if(time_now - packet_check_time >= rtns_check_period)
    {

       cache_media->get_loss_packetno_list(packetno_list);

       //set_rtns_level();

       packet_check_time = time_now;
    }

    return ret;
}

int SrsRtuStreamCaChe::process_rtns_stream_packet(std::vector<DWORD> &video_packetno_list, std::vector<DWORD> &audio_packetno_list)
{
    int ret = ERROR_SUCCESS;
    srs_update_system_time_ms();
    long long time_now = srs_get_system_time_ms();
    if(time_now - packet_check_time >= rtns_check_period)
    {
       cache_media->get_loss_packetno_list(video_packetno_list, audio_packetno_list);
//
      // set_rtns_level();

       packet_check_time = time_now;
    }

    return ret;
}
void SrsRtuStreamCaChe::rtu_to_rtmp()
{
    SrsCommonMessage* msg = 0;
    msg = get_complete_msg();
    if(msg)
    {
//            SrsRtuMsgStruct* msg_struct = new SrsRtuMsgStruct();
//            msg_struct->msg= msg;
//            msg_struct->url = url;
//            _rtu_msg_queue->push_back(msg_struct);
    }
}

SrsCommonMessage*  SrsRtuStreamCaChe::get_complete_msg()
{
    SrsRtuSliceNoMap* sliceno_queue = 0;
    sliceno_queue = get_group_slice_data();


    if(sliceno_queue != 0)
    {
        SrsCommonMessage* msg = 0;
        msg = packets_to_msg(sliceno_queue);

        sliceno_queue->clear();

        srs_freep(sliceno_queue);

        return msg;
    }

    return 0;
}

bool SrsRtuStreamCaChe::get_is_output_flag()
{
    bool output_flag = false;
    if(cache_media->get_is_output_flag())
    {
       output_flag = true;
    }
    else
    {
        output_flag = false;

        /*if(update_output_time_flag)
        {
            output_timeout = rtu_get_system_time();
            update_output_time_flag = false;
        }

        long long time_now = rtu_get_system_time();
        if(time_now - output_timeout > cache_output_time)
        {
            output_flag = true;
            update_output_time_flag = true;
        }*/
    }
    return output_flag;
}


SrsRtuSliceNoMap* SrsRtuStreamCaChe::get_group_slice_data()
{
    SrsRtuSliceNoMap* sliceno_queue = 0;

    if(sh_full_flag)
    {
        if(cache_sh->get_size() > 0)
        {
            if ( cache_sh->check_is_complete() )
            {
                    sliceno_queue= cache_sh->pop();
            }
        }
        else
        {
            sh_full_flag = false;
            process_packet_flag = true;
            first_play_time = rtu_get_system_time();
        }
    }

    if(cache_media->get_size() > 0)
    {
        if(cache_media->get_size() > 600)
        {
                        url.c_str(), cache_media->get_cache_time(), cache_media->get_size(),
                        cache_media->get_insert_payload_size();

        }

        bool del_flag = false;
        if(process_packet_flag)
        {
            if ( cache_media->check_is_complete() )
            {
                if(get_is_output_flag())
                {
                    sliceno_queue= cache_media->pop();

                    force_output_time = rtu_get_system_time();
                }
                else
                {
                    if(cache_media->get_cache_time() >  cache_time )//SRS_RTU_SECONDS_TO_OPEN_CACHE_TIME)
                    {
                        //del_flag = true;
                        sliceno_queue= cache_media->pop();

                        force_output_time = rtu_get_system_time();

                    }
                }
            }
            else
            {
                if(cache_media->get_cache_time() > cache_time)
                {
                    del_flag = true;
                }
            }
        }
        else
        {
            if(cache_media->get_cache_time() >  SRS_RTU_SECONDS_TO_OPEN_CACHE_TIME)
            {
                del_flag = true;
            }
        }

        if(del_flag)
        {
            SrsRtuSliceNoMap* sliceno_queue_del=  cache_media->pop();

            srs_freep(sliceno_queue_del);
        }
    }

    return sliceno_queue;
}

bool SrsRtuStreamCaChe::erase_timeout_incomplete_slice()
{
    bool del_flag = false;

    if(cache_media->get_size() > 0)
    {
        if(process_packet_flag)
        {
            if( cache_media->check_is_complete() == false && cache_media->get_cache_time() > cache_time)
            {
                del_flag = true;
            }
        }
        else
        {
            if(cache_media->get_cache_time() >  SRS_RTU_SECONDS_TO_OPEN_CACHE_TIME)
            {
                del_flag = true;
            }
        }
    }

    if(del_flag)
    {
        SrsRtuSliceNoMap* sliceno_queue_del=  cache_media->pop();

        srs_freep(sliceno_queue_del);
    }

    return del_flag;
}



SrsCommonMessage* SrsRtuStreamCaChe::packets_to_msg(SrsRtuSliceNoMap* sliceno_queue)
{
    SrsCommonMessage* msg = 0;
    SrsRtuRtmpHeader* header = 0;
    SrsRtuStreamDataPacket* pkt = 0;
    int header_size = sizeof(SrsRtuRtmpHeader);

    if(sliceno_queue->get_size() == 0) {
        return 0;
    } else {
        msg = new SrsCommonMessage();
    }

    // first slice get the rtmp message header info
    pkt = sliceno_queue->get_packet(0);//sliceno_queue->pop();

    if(false)//if(pkt->stream_type == SrsRtuStreamDataTypeAudio)
    {
        if(last_audio_packetno != 0)
        {
            if(last_audio_packetno + 1 != pkt->audio_packet_no)
            {
                loss_audio_packet_count++;
                printf("loss packetno = %d ,loss audio packetno = %d, loss_audio_packet_count =%d\n",
                       pkt->packet_no -1, last_audio_packetno + 1, loss_audio_packet_count);
            }
        }
        last_audio_packetno = pkt->audio_packet_no;
    }
    // the rtu packet payload size should be no less than the header_size
   // srs_assert(pkt->payload_size >= header_size);

    if(pkt->payload_size <= 0){
    	return 0;
    }

    // assign the rtmp header info
    header = (SrsRtuRtmpHeader*)pkt->payload();

    msg->header.timestamp_delta = header->timestamp_delta;
    msg->header.payload_length = header->payload_length;
    msg->header.message_type = header->message_type;
    msg->header.stream_id = header->stream_id;
    msg->header.timestamp = header->timestamp;
    msg->header.perfer_cid = header->perfer_cid;
    msg->size = header->size;


    // create the rtmp message payload according to the msg->size
    msg->create_payload(msg->size);

    char* p = msg->payload;
    char* pend = msg->payload + msg->size;

    // copy the payload
    memcpy(msg->payload, pkt->payload() + header_size, pkt->payload_size - header_size);

    // move the current pointer
    p += (pkt->payload_size - header_size);
    srs_assert(p <= pend);

    //srs_freep(pkt);

    //while(sliceno_queue->get_size() != 0) {
    for(int i = 1; i < sliceno_queue->get_slice_sum(); i++) {
        pkt = sliceno_queue->get_packet(i);
        // copy the payload and move pointer
        memcpy(p,pkt->payload(),pkt->payload_size);
        p += pkt->payload_size;
        srs_assert(p <= pend);
        //srs_freep(pkt);
    }

    return msg;
}

int  SrsRtuStreamCaChe::memory_recovery(SrsRtuSliceNoMap* sliceno_map)
{
    return cache_media->memory_recovery(sliceno_map);
}

void SrsRtuStreamCaChe::calculate_max_and_min_packetno(SrsRtuStreamDataPacket* packet)
{
    //calculate max packetno
    if(max_packetno < packet->packet_no && packet->packet_no > SRS_RTU_FIRST_PACKET_NO)
    {
        max_packetno = packet->packet_no;
    }

    //calculate min packetno
    if(packet->packet_no < SRS_RTU_FIRST_PACKET_NO)
    {
        sh_num++;
    }
    else
    {
        if(min_packetno < SRS_RTU_FIRST_PACKET_NO  || min_packetno > packet->packet_no)
        {
            min_packetno = packet->packet_no;
        }
    }
}

void SrsRtuStreamCaChe::calculate_packet_loss_rate()
{
    double denominator = 0.0;
    double fractions = 0.0;
    if(max_packetno < 10)
    {
        denominator = (double)(sh_num );
        fractions = denominator - (double)recv_packet_num;
    }
    else
    {
        denominator = (double)( max_packetno - min_packetno + 1 + sh_num );
        fractions = denominator - (double)recv_packet_num;
    }

    packet_loss_rate = fractions/ denominator;

}


void SrsRtuStreamCaChe::calculate_rtns_packet_loss_rate()
{
    double denominator = 0.0;
    double fractions = 0.0;
    if(max_packetno < 10)
    {
        denominator = (double)(sh_num );
        fractions = denominator - (double)recv_packet_num  + (double)rtns_packet_num;
    }
    else
    {
        denominator = (double)( max_packetno - min_packetno + 1 + sh_num );
        fractions = denominator - (double)recv_packet_num - (double)rtns_packet_num;
    }

    rtns_packet_loss_rate = fractions/ denominator;

}

void SrsRtuStreamCaChe::calculate_video_standstill_rate()
{

}

double SrsRtuStreamCaChe::get_packet_loss_rate()
{
    return packet_loss_rate;
}

double SrsRtuStreamCaChe::get_three_mins_packet_loss_rate()
{
	return 0.0;
}

double SrsRtuStreamCaChe::get_three_mins_audio_packet_loss_rate()
{
	return 0.0;
}

double SrsRtuStreamCaChe::get_three_mins_video_packet_loss_rate()
{
	return 0.0;
}

double SrsRtuStreamCaChe::get_average_packet_loss_rate()
{
	return 0.0;
}

void SrsRtuStreamCaChe::init_cache_param(long long delay_time, int32_t byterate)
{
    if(delay_time <= 60)
    {
        cache_time = 300;  //ms
        rtns_check_period = delay_time + 30; //ms
        rtns_level = 1;
        nagle_cache_time = 200; //ms
        cache_output_waittime = 5;
    }
    else if(delay_time > 60 && delay_time <=100)
    {
        cache_time = 500;  //ms
        rtns_check_period = delay_time + 30; //ms
        rtns_level = 1;
        nagle_cache_time = 400; //ms
        cache_output_waittime = 8;
    }
    else if(delay_time > 100 && delay_time <=200)
    {
        cache_time = 600;  //ms
        rtns_check_period = delay_time + 30; //ms
        rtns_level = 2;
        nagle_cache_time = 500; //ms
        cache_output_waittime = 8;
    }
    else
    {
        cache_time = 400;  //ms
        rtns_check_period = delay_time + 30; //ms
        rtns_level = 3;
        nagle_cache_time = 300; //ms
    }

    if((rtns_check_period * 2) > cache_time)
    {
        cache_output_time = cache_time;
    }
    else
    {
        cache_output_time = rtns_check_period*2;
    }

//    cache_time += SRS_RTU_CACHE_SIZE_RESERVED_TIME;
    cache_time = 200;
    cache_stream_size = byterate * cache_time / 1000;

    nagle_cache_size = cache_stream_size;

    round_trip_time = delay_time;
}

int SrsRtuStreamCaChe::set_rtns_level()
{
	return 0;
}

int SrsRtuStreamCaChe::get_rtns_level()
{
    return rtns_level;
}

